package test.threads.queue.common;

import java.util.concurrent.Callable;

import com.google.common.base.Throwables;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import test.threads.queue.common.FSM.State;

/**
 * @author cheremin
 * @since 24.03.12,  14:41
 */
public abstract class BaseEnqueuer<E> implements Runnable, Callable<Void> {
    private static final Log log = LogFactory.getLog( BaseEnqueuer.class );

    private final FSM fsm = new FSM();

    private final int eventsInBatch;

    private volatile long eventsEnqueued = 0;

    public BaseEnqueuer( final int eventsInBatch ) {
        this.eventsInBatch = eventsInBatch;
    }

    @Override
    public Void call() throws Exception {
        run();
        return null;
    }

    @Override
    public void run() {
        try {
            fsm.initialize();
            do {
                final State state = fsm.autoTransition();
                switch ( state ) {
                    case WAITING:
                        waiting();
                        break;
                    case RUNNING:
                        running();
                        break;
                    case TERMINATING:
                        log.info( getClass().getSimpleName() + " terminating" );
                        return;
                    default:
                        throw new IllegalStateException( "Unexpected state: " + state );

                }
            } while ( true );
        } catch ( InterruptedException e ) {
            log.info( getClass().getSimpleName() + " interrupted" );
        } catch ( Throwable t ) {
            log.error( getClass().getSimpleName() + " exited unexpectedly", t );
            throw Throwables.propagate( t );
        }
    }

    private void waiting() throws InterruptedException {
        while ( fsm.is( State.WAITING ) ) {
            Thread.yield();
            if ( Thread.interrupted() ) {
                throw new InterruptedException();
            }
        }
    }

    private void running() throws Exception {
        eventsEnqueued = 0;
        while ( fsm.is( State.RUNNING ) ) {
            enqueueEventsBatch( eventsInBatch );
            eventsEnqueued += eventsInBatch;
        }
    }

    protected abstract void enqueueEventsBatch( final int eventsInBatch ) throws Exception;

    public void waitForInitialized() {
        while ( !fsm.is( State.WAITING ) ) {
            Thread.yield();
        }
    }

    public void start() {
        fsm.start();
    }

    public void waitForStarting() {
        fsm.waitForState( State.RUNNING );
    }

    public void pause() {
        fsm.pause();
    }

    public void waitForPausing() {
        fsm.waitForState( State.WAITING );
    }

    public boolean is( final State expected ) {
        return fsm.is( expected );
    }

    public void terminate() {
        fsm.terminate();
    }

    public long eventsEnqueued() {
        return eventsEnqueued;
    }
}
